from typing import Dict, Any, Optional, List, Callable
from datetime import datetime, timedelta
import asyncio
import json
from dataclasses import dataclass, asdict
from enum import Enum
import pybreaker
from middleware.redis_adapter import get_redis_adapter
from tenacity import retry, stop_after_attempt, wait_exponential
import threading
import time

from utils.logger import Logger
from utils.exceptions import CircuitBreakerOpenError, MaessBaseException
from config import agent_config

# 自定义缺失的异常类
class CircuitBreakerError(MaessBaseException):
    """熔断器相关错误"""
    def __init__(self, message: Optional[str] = None, context: Optional[Dict[str, Any]] = None):
        super().__init__(
            error_code=getattr("ErrorCode", "CIRCUIT_BREAKER_ERROR", 50001),
            message=message or "Circuit breaker error",
            context=context
        )

class DistributedLockError(MaessBaseException):
    """分布式锁错误"""
    def __init__(self, message: Optional[str] = None, context: Optional[Dict[str, Any]] = None):
        super().__init__(
            error_code=getattr("ErrorCode", "LOCK_ERROR", 50002),
            message=message or "Distributed lock error",
            context=context
        )

logger = Logger.get_logger("circuit_breaker_manager")


class CircuitState(Enum):
    """熔断器状态"""
    CLOSED = "closed"      # 正常状态
    OPEN = "open"          # 熔断状态
    HALF_OPEN = "half_open"  # 半开状态


# 自定义熔断器实现（支持异步操作）
class CustomCircuitBreaker:
    """自定义熔断器核心类（每个依赖/Agent 对应一个实例）"""

    def __init__(
            self,
            name: str,
            failure_threshold: int = 5,
            recovery_timeout: int = 30,
            half_open_test_count: int = 3
    ):
        self.name = name  # 熔断器名称
        self.state: CircuitState = CircuitState.CLOSED  # 初始状态
        self.failure_count: int = 0  # 连续失败次数
        self.success_count: int = 0  # 半打开状态下的成功次数
        self.last_failure_time: Optional[datetime] = None  # 最后一次失败时间
        # 配置参数
        self.failure_threshold = failure_threshold  # 触发熔断的失败阈值
        self.recovery_timeout = recovery_timeout  # 熔断恢复时间（秒）
        self.half_open_test_count = half_open_test_count  # 半打开试探次数
        self._lock = asyncio.Lock()  # 异步锁，保证线程安全
        logger.info(f"自定义熔断器初始化: name={name}, threshold={failure_threshold}, recovery={recovery_timeout}s")

    async def _transition_to_closed(self) -> None:
        """切换到关闭状态（服务恢复正常）"""
        async with self._lock:
            self.state = CircuitState.CLOSED
            self.failure_count = 0
            self.success_count = 0
            logger.info(f"熔断器切换到关闭状态: name={self.name}")

    async def _transition_to_open(self) -> None:
        """切换到打开状态（触发熔断）"""
        async with self._lock:
            self.state = CircuitState.OPEN
            self.last_failure_time = datetime.now()
            logger.warning(f"熔断器触发熔断（打开状态）: name={self.name}, 失败次数={self.failure_count}")

    async def _transition_to_half_open(self) -> None:
        """切换到半打开状态（试探服务是否恢复）"""
        async with self._lock:
            self.state = CircuitState.HALF_OPEN
            self.success_count = 0
            logger.info(f"熔断器切换到半打开状态: name={self.name}")

    async def _check_recovery_timeout(self) -> bool:
        """检查熔断恢复时间是否到期"""
        if not self.last_failure_time:
            return False
        return (datetime.now() - self.last_failure_time) >= timedelta(seconds=self.recovery_timeout)

    async def call_async(self, func: Callable, *args, **kwargs) -> Any:
        """异步调用被保护函数"""
        # 1. 检查当前状态
        if self.state == CircuitState.OPEN:
            # 打开状态：检查是否到恢复时间
            if await self._check_recovery_timeout():
                await self._transition_to_half_open()
            else:
                # 未到恢复时间，直接抛出熔断异常
                raise CircuitBreakerOpenError(
                    message=f"熔断器已打开，拒绝请求（{self.recovery_timeout}秒后重试）",
                    context={"breaker_name": self.name, "state": self.state.value}
                )

        # 2. 执行目标函数
        try:
            result = await func(*args, **kwargs)
            # 3. 处理成功逻辑
            if self.state == CircuitState.HALF_OPEN:
                async with self._lock:
                    self.success_count += 1
                    logger.debug(
                        f"半打开状态请求成功: name={self.name}, 成功次数={self.success_count}/{self.half_open_test_count}")
                    # 试探成功次数达标，切换到关闭状态
                    if self.success_count >= self.half_open_test_count:
                        await self._transition_to_closed()
            elif self.state == CircuitState.CLOSED:
                # 关闭状态下，重置失败计数
                async with self._lock:
                    self.failure_count = 0
            return result
        except Exception as e:
            # 4. 处理失败逻辑
            async with self._lock:
                self.failure_count += 1
                logger.warning(f"请求失败: name={self.name}, 连续失败次数={self.failure_count}, error={str(e)[:50]}")

                if self.state == CircuitState.CLOSED:
                    # 关闭状态：失败次数达标，切换到打开状态
                    if self.failure_count >= self.failure_threshold:
                        await self._transition_to_open()
                elif self.state == CircuitState.HALF_OPEN:
                    # 半打开状态：任何失败都切换回打开状态
                    await self._transition_to_open()

            # 抛出原异常
            raise

    def call(self, func: Callable, *args, **kwargs) -> Any:
        """同步调用被保护函数"""
        # 1. 检查当前状态
        if self.state == CircuitState.OPEN:
            # 打开状态：检查是否到恢复时间
            if (datetime.now() - self.last_failure_time) >= timedelta(seconds=self.recovery_timeout):
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
                logger.info(f"熔断器切换到半打开状态: name={self.name}")
            else:
                # 未到恢复时间，直接抛出熔断异常
                raise CircuitBreakerOpenError(
                    message=f"熔断器已打开，拒绝请求（{self.recovery_timeout}秒后重试）",
                    context={"breaker_name": self.name, "state": self.state.value}
                )

        # 2. 执行目标函数
        try:
            result = func(*args, **kwargs)
            # 3. 处理成功逻辑
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                logger.debug(
                    f"半打开状态请求成功: name={self.name}, 成功次数={self.success_count}/{self.half_open_test_count}")
                # 试探成功次数达标，切换到关闭状态
                if self.success_count >= self.half_open_test_count:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
                    logger.info(f"熔断器切换到关闭状态: name={self.name}")
            elif self.state == CircuitState.CLOSED:
                # 关闭状态下，重置失败计数
                self.failure_count = 0
            return result
        except Exception as e:
            # 4. 处理失败逻辑
            self.failure_count += 1
            logger.warning(f"请求失败: name={self.name}, 连续失败次数={self.failure_count}, error={str(e)[:50]}")

            if self.state == CircuitState.CLOSED:
                # 关闭状态：失败次数达标，切换到打开状态
                if self.failure_count >= self.failure_threshold:
                    self.state = CircuitState.OPEN
                    self.last_failure_time = datetime.now()
                    logger.warning(f"熔断器触发熔断（打开状态）: name={self.name}, 失败次数={self.failure_count}")
            elif self.state == CircuitState.HALF_OPEN:
                # 半打开状态：任何失败都切换回打开状态
                self.state = CircuitState.OPEN
                self.last_failure_time = datetime.now()
                logger.warning(f"熔断器触发熔断（打开状态）: name={self.name}, 失败次数={self.failure_count}")

            # 抛出原异常
            raise

    def get_state(self) -> Dict[str, Any]:
        """获取熔断器当前状态"""
        return {
            "name": self.name,
            "state": self.state.value,
            "failure_count": self.failure_count,
            "success_count": self.success_count,
            "last_failure_time": self.last_failure_time.strftime(
                "%Y-%m-%d %H:%M:%S") if self.last_failure_time else None,
            "recovery_timeout": self.recovery_timeout,
            "failure_threshold": self.failure_threshold
        }


@dataclass
class CircuitBreakerConfig:
    """熔断器配置"""
    failure_threshold: int = 5          # 失败阈值
    recovery_timeout: int = 60          # 恢复超时时间（秒）
    expected_exception: type = Exception  # 预期异常类型
    name: str = "default"               # 熔断器名称
    enable_logging: bool = True           # 启用日志记录


@dataclass
class DistributedLockConfig:
    """分布式锁配置"""
    ttl_seconds: int = 30               # 锁的TTL（秒）
    retry_count: int = 3                # 重试次数
    retry_delay: float = 0.1            # 重试延迟（秒）
    lock_key_prefix: str = "maess_lock"   # 锁键前缀


class CircuitBreakerManager:
    """熔断器管理器（同时支持Redis分布式和本地模式）"""

    def __init__(self, redis_url: str = None):
        self.redis_client = get_redis_adapter()  # 使用Redis适配器，支持集群模式
        self.circuit_breakers: Dict[str, Any] = {}  # 可以是pybreaker或CustomCircuitBreaker
        self.use_redis = True  # 默认使用Redis模式
        self._initialize_redis()

    def _initialize_redis(self):
        """初始化Redis连接"""
        try:
            # 测试连接
            self.redis_client.ping()
            logger.info("熔断器Redis连接初始化完成")
            self.redis_connected = True
        except Exception as e:
            logger.error(f"熔断器Redis连接失败: {str(e)}")
            self.redis_connected = False
            self.use_redis = False  # 自动切换到本地模式
            logger.warning("熔断器自动降级到本地模式运行")
    
    async def async_call_with_circuit_breaker(self, name: str, func: Callable, *args, **kwargs):
        """异步方式调用熔断器包装的函数（使用统一的异步工具）"""
        from utils.async_utils import safe_async_call, run_sync_in_background
        
        breaker = self.get_circuit_breaker(name)
        
        # 根据熔断器类型执行不同的调用方式
        if isinstance(breaker, pybreaker.CircuitBreaker):
            # pybreaker异步调用使用线程池
            try:
                return await run_sync_in_background(
                    breaker.call,
                    lambda: func(*args, **kwargs)
                )
            except pybreaker.CircuitBreakerError as e:
                logger.error(f"熔断器打开，阻止请求: {name}, {str(e)}")
                raise CircuitBreakerOpenError(f"熔断器 {name} 已打开")
        elif hasattr(breaker, 'call_async'):
            # 自定义熔断器异步调用，添加安全包装
            return await safe_async_call(breaker.call_async, func, *args, **kwargs)
        else:
            # 降级调用
            return await safe_async_call(func, *args, **kwargs)

    def create_circuit_breaker(self, config: CircuitBreakerConfig) -> Any:
        """创建熔断器（支持自动降级）"""
        try:
            # 尝试从agent配置获取特定参数
            agent_conf = None
            try:
                agent_conf = agent_config.get_agent_config(config.name)
            except Exception:
                pass
            
            # 优先使用Redis分布式模式（如果可用）
            if self.use_redis:
                try:
                    # 验证Redis连接
                    self.redis_client.ping()
                    
                    # 创建Redis状态存储
                    redis_store = pybreaker.CircuitRedisStorage(
                        state=pybreaker.STATE_CLOSED,
                        redis_object=self.redis_client,
                        namespace=f"circuit_breaker:{config.name}"
                    )
                    
                    # 创建熔断器
                    failure_threshold = agent_conf.max_concurrent // 10 if agent_conf else config.failure_threshold
                    circuit_breaker = pybreaker.CircuitBreaker(
                        fail_max=failure_threshold,
                        reset_timeout=config.recovery_timeout,
                        exclude=[config.expected_exception],
                        state_storage=redis_store,
                        name=config.name
                    )
                    
                    # 添加监听器
                    if config.enable_logging:
                        circuit_breaker.add_listener(CircuitBreakerListener(config.name))
                    
                    self.circuit_breakers[config.name] = circuit_breaker
                    logger.info(f"Redis分布式熔断器创建完成: {config.name}")
                    return circuit_breaker
                except Exception as redis_error:
                    logger.warning(f"Redis熔断器创建失败，降级到本地模式: {str(redis_error)}")
                    self.use_redis = False  # 标记Redis不可用
            
            # 降级到本地自定义熔断器
            failure_threshold = agent_conf.max_concurrent // 10 if agent_conf else config.failure_threshold
            circuit_breaker = CustomCircuitBreaker(
                name=config.name,
                failure_threshold=failure_threshold,
                recovery_timeout=config.recovery_timeout,
                half_open_test_count=3
            )
            
            self.circuit_breakers[config.name] = circuit_breaker
            logger.info(f"本地自定义熔断器创建完成: {config.name}")
            return circuit_breaker
            
        except Exception as e:
            logger.error(f"创建熔断器失败: {str(e)}")
            raise CircuitBreakerError(f"创建熔断器失败: {str(e)}")

    def get_circuit_breaker(self, name: str) -> Any:
        """获取熔断器（支持两种类型）"""
        if name not in self.circuit_breakers:
            # 如果熔断器不存在，创建默认配置的熔断器
            default_config = CircuitBreakerConfig(
                name=name,
                failure_threshold=3,
                recovery_timeout=60,
                expected_exception=Exception,
                enable_logging=True
            )
            return self.create_circuit_breaker(default_config)
        return self.circuit_breakers[name]

    def get_circuit_state(self, name: str) -> Optional[CircuitState]:
        """获取熔断器状态"""
        circuit_breaker = self.get_circuit_breaker(name)
        if not circuit_breaker:
            return None
        
        if hasattr(circuit_breaker, 'get_state'):
            # 自定义熔断器获取状态方式
            state_value = circuit_breaker.get_state().get('state')
            for state in CircuitState:
                if state.value == state_value:
                    return state
        else:
            # pybreaker熔断器状态映射
            state_map = {
                pybreaker.STATE_CLOSED: CircuitState.CLOSED,
                pybreaker.STATE_OPEN: CircuitState.OPEN,
                pybreaker.STATE_HALF_OPEN: CircuitState.HALF_OPEN
            }
            
            return state_map.get(circuit_breaker.current_state)
        return None

    def reset_circuit_breaker(self, name: str) -> bool:
        """重置熔断器（支持两种类型）"""
        try:
            circuit_breaker = self.get_circuit_breaker(name)
            if not circuit_breaker:
                return False
            
            # 根据不同类型的熔断器执行不同的重置逻辑
            if isinstance(circuit_breaker, pybreaker.CircuitBreaker):
                circuit_breaker.close()
            elif hasattr(circuit_breaker, '_transition_to_closed'):
                # 自定义熔断器的重置方法
                asyncio.run(circuit_breaker._transition_to_closed())
            
            logger.info(f"熔断器重置完成: {name}")
            return True
            
        except Exception as e:
            logger.error(f"重置熔断器失败: {str(e)}")
            return False

    def get_all_circuit_states(self) -> Dict[str, CircuitState]:
        """获取所有熔断器状态"""
        states = {}
        for name, circuit_breaker in self.circuit_breakers.items():
            state = self.get_circuit_state(name)
            if state:
                states[name] = state
        return states


class CircuitBreakerListener(pybreaker.CircuitBreakerListener):
    """熔断器监听器"""

    def __init__(self, name: str):
        self.name = name

    def state_change(self, cb, old_state, new_state):
        """状态变更监听"""
        logger.info(f"熔断器状态变更: {self.name}, {old_state} -> {new_state}")

    def failure(self, cb, exc):
        """失败监听"""
        logger.warning(f"熔断器失败: {self.name}, 异常: {str(exc)}")

    def success(self, cb):
        """成功监听"""
        logger.info(f"熔断器成功: {self.name}")


class DistributedLockManager:
    """分布式锁管理器"""

    def __init__(self, redis_url: str = None, 
                 config: Optional[DistributedLockConfig] = None):
        self.redis_client = get_redis_adapter()  # 使用Redis适配器，支持集群模式
        self.config = config or DistributedLockConfig()
        self.redlock_factory = None
        self._initialize_redis()
        
    def _init_redlock_factory(self):
        """初始化RedLock工厂"""
        try:
            # 尝试导入redlock库
            import redlock
            # 初始化RedLock工厂
            self.redlock_factory = redlock.RedLockFactory([self.redis_client])
            return True
        except ImportError:
            logger.warning("redlock库未安装，分布式锁功能将受限")
            return False
        except Exception as e:
            logger.error(f"初始化RedLock工厂失败: {str(e)}")
            return False

    def _initialize_redis(self):
        """初始化Redis连接"""
        try:
            # 测试连接
            self.redis_client.ping()
            logger.info("分布式锁Redis连接初始化完成")
            
            # 尝试初始化redlock工厂
            self._init_redlock_factory()
            
        except Exception as e:
            logger.error(f"分布式锁Redis连接失败: {str(e)}")
            raise DistributedLockError(f"分布式锁Redis连接失败: {str(e)}")
            
    async def _acquire_redis_simple_lock(self, key: str, ttl: int) -> bool:
        """使用Redis的SETNX实现简单锁"""
        try:
            return self.redis_client.set(key, 1, nx=True, ex=ttl)
        except Exception as e:
            logger.error(f"获取简单Redis锁失败: {str(e)}")
            return False
            
    async def _release_redis_simple_lock(self, key: str) -> bool:
        """释放简单Redis锁"""
        try:
            return self.redis_client.delete(key) > 0
        except Exception as e:
            logger.error(f"释放简单Redis锁失败: {str(e)}")
            return False
            
    async def _check_redis_simple_lock(self, key: str) -> bool:
        """检查简单Redis锁是否存在"""
        try:
            return self.redis_client.exists(key) > 0
        except Exception as e:
            logger.error(f"检查简单Redis锁失败: {str(e)}")
            return True  # 默认认为锁被占用

    async def acquire_lock(self, lock_key: str, ttl_seconds: Optional[int] = None,
                          retry_count: Optional[int] = None, 
                          retry_delay: Optional[float] = None):
        """获取分布式锁"""
        """获取分布式锁"""
        try:
            ttl = ttl_seconds or self.config.ttl_seconds
            retry_count_val = retry_count or self.config.retry_count
            retry_delay_val = retry_delay or self.config.retry_delay
            
            # 构建完整的锁键
            full_lock_key = f"{self.config.lock_key_prefix}:{lock_key}"
            
            # 尝试获取锁
            for attempt in range(retry_count_val + 1):
                try:
                    # 如果redlock_factory未初始化，尝试初始化
                    if not self.redlock_factory:
                        if not self._init_redlock_factory():
                            # 降级方案：使用Redis的SETNX实现简单锁
                            if await self._acquire_redis_simple_lock(full_lock_key, ttl):
                                logger.info(f"降级方案: 分布式锁获取成功: {full_lock_key}")
                                return {'key': full_lock_key, 'ttl': ttl}
                            continue
                    
                    # 使用RedLock获取分布式锁
                    lock = self.redlock_factory.create_lock(
                        resource=full_lock_key.encode('utf-8'),
                        ttl=ttl * 1000  # 转换为毫秒
                    )
                    
                    if lock.acquire():
                        logger.info(f"分布式锁获取成功: {full_lock_key}")
                        return lock
                    
                except Exception as e:
                    logger.warning(f"分布式锁获取尝试失败: {full_lock_key}, 尝试 {attempt + 1}, 错误: {str(e)}")
                
                # 如果不是最后一次尝试，等待后重试
                if attempt < retry_count_val:
                    await asyncio.sleep(retry_delay_val * (2 ** attempt))  # 指数退避
            
            logger.error(f"分布式锁获取失败: {full_lock_key}, 重试 {retry_count_val + 1} 次后失败")
            return None
            
        except Exception as e:
            logger.error(f"获取分布式锁失败: {str(e)}")
            raise DistributedLockError(f"获取分布式锁失败: {str(e)}")

    async def release_lock(self, lock) -> bool:
        """释放分布式锁"""
        try:
            # 检查是否为字典格式的简单锁
            if isinstance(lock, dict) and 'key' in lock:
                # 释放简单锁
                await self._release_redis_simple_lock(lock['key'])
                logger.info(f"降级方案: 分布式锁释放成功: {lock['key']}")
                return True
            # 检查是否为RedLock实例
            elif hasattr(lock, 'locked') and hasattr(lock, 'release') and lock and lock.locked():
                lock.release()
                logger.info(f"分布式锁释放成功")
                return True
            else:
                logger.warning(f"分布式锁释放失败: 锁不存在或未锁定")
                return False
                
        except Exception as e:
            logger.error(f"释放分布式锁失败: {str(e)}")
            return False

    async def is_locked(self, lock_key: str) -> bool:
        """检查锁是否被占用"""
        try:
            full_lock_key = f"{self.config.lock_key_prefix}:{lock_key}"
            
            # 如果redlock_factory未初始化，使用简单锁检查
            if not self.redlock_factory:
                # 检查简单锁是否存在
                exists = await self._check_redis_simple_lock(full_lock_key)
                return exists
            
            # 尝试获取锁（不阻塞）
            lock = self.redlock_factory.create_lock(
                resource=full_lock_key.encode('utf-8'),
                ttl=1000  # 1秒TTL
            )
            
            # 尝试非阻塞获取
            if lock.acquire(blocking=False):
                lock.release()  # 立即释放
                return False  # 锁可用
            else:
                return True  # 锁被占用
                
        except Exception as e:
            logger.error(f"检查锁状态失败: {str(e)}")
            return True  # 默认认为锁被占用

    async def with_distributed_lock(self, lock_key: str, 
                                   func, *args, **kwargs):
        """使用分布式锁执行函数"""
        lock = None
        try:
            # 获取锁
            lock = await self.acquire_lock(lock_key)
            if not lock:
                raise DistributedLockError(f"无法获取分布式锁: {lock_key}")
            
            # 执行函数
            if asyncio.iscoroutinefunction(func):
                result = await func(*args, **kwargs)
            else:
                result = func(*args, **kwargs)
            
            return result
            
        finally:
            # 释放锁
            if lock:
                await self.release_lock(lock)

    def get_lock_info(self, lock_key: str) -> Dict[str, Any]:
        """获取锁信息"""
        try:
            full_lock_key = f"{self.config.lock_key_prefix}:{lock_key}"
            
            # 检查锁状态
            is_locked = self.is_locked(lock_key)
            
            # 获取锁的TTL（如果存在）
            ttl = None
            try:
                ttl = self.redis_client.ttl(full_lock_key)
            except Exception:
                ttl = -1
            
            return {
                "lock_key": full_lock_key,
                "is_locked": is_locked,
                "ttl_seconds": ttl if ttl and ttl > 0 else None,
                "timestamp": datetime.now().isoformat()
            }
            
        except Exception as e:
            logger.error(f"获取锁信息失败: {str(e)}")
            return {
                "lock_key": lock_key,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

    async def cleanup_expired_locks(self) -> int:
        """清理过期锁"""
        try:
            cleaned_count = 0
            
            # 扫描所有锁键
            cursor = 0
            while True:
                cursor, keys = self.redis_client.scan(
                    cursor, match=f"{self.config.lock_key_prefix}:*", count=100
                )
                
                for key in keys:
                    try:
                        # 检查TTL
                        ttl = self.redis_client.ttl(key)
                        if ttl == -2:  # 键不存在
                            cleaned_count += 1
                        elif ttl == -1:  # 没有设置过期时间
                            # 可以设置一个默认过期时间
                            self.redis_client.expire(key, self.config.ttl_seconds)
                    except Exception as e:
                        logger.warning(f"清理锁失败: {key}, 错误: {str(e)}")
                
                if cursor == 0:
                    break
            
            logger.info(f"过期锁清理完成: 清理了 {cleaned_count} 个锁")
            return cleaned_count
            
        except Exception as e:
            logger.error(f"清理过期锁失败: {str(e)}")
            return 0

    async def health_check(self) -> Dict[str, Any]:
        """健康检查"""
        try:
            # 测试Redis连接
            redis_ok = self.redis_client.ping()
            
            # 测试锁功能
            test_lock_key = "health_check_test"
            lock = await self.acquire_lock(test_lock_key, ttl_seconds=5)
            lock_ok = lock is not None
            
            if lock:
                await self.release_lock(lock)
            
            return {
                "status": "healthy" if redis_ok and lock_ok else "unhealthy",
                "redis": {"status": "connected" if redis_ok else "disconnected"},
                "lock_function": {"status": "working" if lock_ok else "failed"},
                "timestamp": datetime.now().isoformat()
            }
            
        except Exception as e:
            logger.error(f"健康检查失败: {str(e)}")
            return {
                "status": "error",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

    async def close(self):
        """关闭连接"""
        try:
            if self.redis_client:
                self.redis_client.close()
                logger.info("分布式锁Redis连接已关闭")
        except Exception as e:
            logger.error(f"关闭连接失败: {str(e)}")


# 装饰器：使用熔断器保护
def with_circuit_breaker(circuit_breaker_name: str = "default"):
    """熔断器装饰器"""
    def decorator(func):
        @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
        async def async_wrapper(*args, **kwargs):
            circuit_breaker_manager = kwargs.get('_circuit_breaker_manager')
            if not circuit_breaker_manager:
                # 使用延迟导入避免循环依赖
                from states.circuit_breaker_manager import get_circuit_breaker_manager
                circuit_breaker_manager = await get_circuit_breaker_manager()
            
            circuit_breaker = circuit_breaker_manager.get_circuit_breaker(circuit_breaker_name)
            if not circuit_breaker:
                # 创建默认熔断器
                config = CircuitBreakerConfig(name=circuit_breaker_name)
                circuit_breaker = circuit_breaker_manager.create_circuit_breaker(config)
            
            # 使用熔断器执行函数
            if asyncio.iscoroutinefunction(func):
                return await circuit_breaker.call_async(func, *args, **kwargs)
            else:
                # 如果是同步函数但在异步环境中调用，使用线程池执行
                loop = asyncio.get_event_loop()
                return await loop.run_in_executor(
                    None, 
                    lambda: circuit_breaker.call(func, *args, **kwargs)
                )
        
        def sync_wrapper(*args, **kwargs):
            circuit_breaker_manager = kwargs.get('_circuit_breaker_manager')
            if not circuit_breaker_manager:
                # 同步环境中导入
                from states.circuit_breaker_manager import get_circuit_breaker_manager
                # 在同步环境中，我们需要运行事件循环来获取管理器实例
                try:
                    # 尝试获取当前事件循环
                    loop = asyncio.get_event_loop()
                    if loop.is_running():
                        # 如果循环正在运行，我们不能调用run_until_complete
                        # 使用降级方案
                        circuit_breaker_manager = None
                    else:
                        circuit_breaker_manager = loop.run_until_complete(get_circuit_breaker_manager())
                except RuntimeError:
                    # 没有运行的事件循环，创建一个新的
                    loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(loop)
                    circuit_breaker_manager = loop.run_until_complete(get_circuit_breaker_manager())
                    loop.close()
            
            if not circuit_breaker_manager:
                # 降级方案：直接执行函数
                logger.warning(f"熔断器管理器不可用，直接执行函数: {circuit_breaker_name}")
                return func(*args, **kwargs)
            
            circuit_breaker = circuit_breaker_manager.get_circuit_breaker(circuit_breaker_name)
            if not circuit_breaker:
                # 创建默认熔断器
                config = CircuitBreakerConfig(name=circuit_breaker_name)
                circuit_breaker = circuit_breaker_manager.create_circuit_breaker(config)
            
            return circuit_breaker.call(func, *args, **kwargs)
        
        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
    
    return decorator


# 全局管理器实例
circuit_breaker_manager = None
distributed_lock_manager = None


async def get_circuit_breaker_manager() -> CircuitBreakerManager:
    """获取熔断器管理器实例"""
    global circuit_breaker_manager
    if circuit_breaker_manager is None:
        try:
            # 确保Redis适配器可用
            from middleware.redis_adapter import get_redis_adapter
            # 检查Redis连接
            redis_client = get_redis_adapter()
            try:
                redis_client.ping()
                logger.info("Redis连接测试成功")
            except Exception as e:
                logger.warning(f"Redis连接测试失败，使用降级模式: {str(e)}")
            
            circuit_breaker_manager = CircuitBreakerManager()
            
            # 创建默认熔断器
            default_configs = [
                CircuitBreakerConfig("text_retrieval", failure_threshold=5, recovery_timeout=60),
                CircuitBreakerConfig("doc_segment", failure_threshold=3, recovery_timeout=120),
                CircuitBreakerConfig("multi_modal", failure_threshold=4, recovery_timeout=90),
                CircuitBreakerConfig("validation", failure_threshold=2, recovery_timeout=30),
                CircuitBreakerConfig("integration", failure_threshold=3, recovery_timeout=60),
                CircuitBreakerConfig("crawler", failure_threshold=5, recovery_timeout=180),
                CircuitBreakerConfig("async_text_qa_agent", failure_threshold=5, recovery_timeout=60),  # 为Text QA Agent添加专用熔断器
            ]
            
            for config in default_configs:
                try:
                    circuit_breaker_manager.create_circuit_breaker(config)
                except Exception as e:
                    logger.error(f"创建熔断器 {config.name} 失败: {str(e)}")
                    # 继续创建其他熔断器
        except Exception as e:
            logger.error(f"初始化熔断器管理器失败: {str(e)}")
            # 创建降级版本的熔断器管理器
            class FallbackCircuitBreakerManager:
                def get_circuit_breaker(self, name):
                    return None
                def create_circuit_breaker(self, config):
                    logger.warning(f"使用降级熔断器: {config.name}")
                    # 返回一个模拟的熔断器对象
                    class MockCircuitBreaker:
                        def __init__(self):
                            self.current_state = 0  # 模拟CLOSED状态
                        async def call_async(self, func, *args, **kwargs):
                            return await func(*args, **kwargs)
                        def call(self, func, *args, **kwargs):
                            return func(*args, **kwargs)
                    return MockCircuitBreaker()
            circuit_breaker_manager = FallbackCircuitBreakerManager()
    
    return circuit_breaker_manager


async def get_distributed_lock_manager() -> DistributedLockManager:
    """获取分布式锁管理器实例"""
    global distributed_lock_manager
    if distributed_lock_manager is None:
        distributed_lock_manager = DistributedLockManager()
    return distributed_lock_manager